package com.bridgeintelligent.tag.bulkload.service.es;

import com.bridgeintelligent.tag.bulkload.api.BulkLoadConfig;
import com.bridgeintelligent.tag.bulkload.api.BulkLoadException;
import com.bridgeintelligent.tag.bulkload.api.fileload.FileLoadModel;
import com.bridgeintelligent.tag.bulkload.service.core.FileService;
import com.bridgeintelligent.tag.bulkload.service.es.index.MappingHelper;
import com.bridgeintelligent.tag.bulkload.service.es.query.IQueryCallback;
import com.wayneleo.quickstart.framework.ExceptionCode;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.StopWatch;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.admin.indices.mapping.put.PutMappingRequest;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.get.*;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.ClearScrollRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchScrollRequest;
import org.elasticsearch.action.support.replication.ReplicationResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.GetMappingsRequest;
import org.elasticsearch.client.indices.GetMappingsResponse;
import org.elasticsearch.cluster.metadata.MappingMetadata;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.core.TimeValue;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.UpdateByQueryRequest;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.script.Script;
import org.elasticsearch.search.Scroll;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
import org.elasticsearch.search.slice.SliceBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;

/**
 * Created by JackCheung on 2019/9/7 11:08 PM.
 */
@Slf4j
public class ESHelper implements IESHelper {

    private RestHighLevelClient client;
    private ESConfig esConfig;
    private static Logger logger = LoggerFactory.getLogger(ESHelper.class);
    private FileService fileService;

    @Override
    public RestHighLevelClient getClient() {
        return client;
    }
    /**
     * 配置ES
     *
     * @param esConfig
     */
    ESHelper(ESConfig esConfig, FileService fileService) {
        this.fileService = fileService;
        String[] address = esConfig.getAddress().split(";");
        HttpHost[] httpHosts = new HttpHost[address.length];
        for (int i = 0; i < address.length; i++) {
            httpHosts[i] = new HttpHost(StringUtils.substringBefore(address[i], ":"),
                                        Integer.parseInt(StringUtils.substringAfter(address[i], ":")),
                                        "http");
        }
        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY,
                                           new UsernamePasswordCredentials(esConfig.getUsername(),
                                                                           esConfig.getPassword()));
        client = new RestHighLevelClient(RestClient.builder(httpHosts)
                                                   .setHttpClientConfigCallback(httpAsyncClientBuilder -> httpAsyncClientBuilder
                                                           .setDefaultCredentialsProvider(credentialsProvider)
                                                           .setKeepAliveStrategy((httpResponse, httpContext) -> TimeUnit.MINUTES
                                                                   .toMinutes(6)))
                                                   .setRequestConfigCallback(requestConfigCallback -> requestConfigCallback
                                                           .setConnectTimeout(600000)
                                                           .setSocketTimeout(600000)));
        this.esConfig = esConfig;
    }

    /**
     * 创建索引
     *
     * @param indexName
     */
    @Override
    public void createIndex(String indexName, String indexType) {
        CreateIndexRequest request = new CreateIndexRequest(indexName);
        request.settings(Settings.builder()
                                 .put("index.number_of_shards", esConfig.getShards())
                                 .put("index.number_of_replicas", esConfig.getReplicas())
                                 .put("max_result_window", esConfig.getMaxWindow())
                                 .put("analysis.analyzer.default.tokenizer", "standard"));
        request.mapping(indexType, MappingHelper.createMapping(indexType));
        try {
            client.indices().create(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public Set<String> getIndexMapping(String indexName)  {
        GetMappingsRequest request = new GetMappingsRequest().indices(indexName);
        GetMappingsResponse response;
        try {
            response = client.indices().getMapping(request, RequestOptions.DEFAULT);
            Map<String, MappingMetadata> mappings = response.mappings();
            MappingMetadata indexMapping = mappings.get(indexName);
            Map<String, Object> mapping = indexMapping.sourceAsMap();
            Map<String,Object> properties = (Map<String,Object> )mapping.get("properties");
            return properties.keySet();
        }catch (IOException e){
            e.printStackTrace();
            return new HashSet<>();
        }
    }

    /**
     * 设置索引字段映射
     *
     * @param indexName
     * @param map
     */
    @Override
    public void updateIndexMapping(String indexName, String indexType, Map<String, String> map) {
        PutMappingRequest request = new PutMappingRequest(indexName);
        request.type(indexType);//类型
        request.source(MappingHelper.updateMapping(map));
        try {
            client.indices().putMapping(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 删除索引
     *
     * @param indexName
     */
    @Override
    public void delIndex(String indexName) {
        DeleteIndexRequest request = new DeleteIndexRequest(indexName);
        try {
            client.indices().delete(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 新增文档
     *
     * @param indexName
     * @param documentId
     * @param map
     */
    @Override
    public void insert(String indexName, String indexType, String documentId, Map<String, Object> map) {
        IndexRequest indexRequest = new IndexRequest(indexName, indexType, documentId).source(map);
        IndexResponse indexResponse = null;
        try {
            indexResponse = client.index(indexRequest, RequestOptions.DEFAULT);
        } catch (ElasticsearchException e) {
            if (e.status() == RestStatus.CONFLICT) {
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        if (indexResponse != null) {
            if (indexResponse.getResult() == DocWriteResponse.Result.CREATED) {
                logger.debug("新增文档成功");
            } else if (indexResponse.getResult() == DocWriteResponse.Result.UPDATED) {
                logger.debug("修改文档成功");
            }
            ReplicationResponse.ShardInfo shardInfo = indexResponse.getShardInfo();
            if (shardInfo.getFailed() > 0) {
                for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                    String reason = failure.reason();
                    logger.error("副本失败原因：" + reason);
                }
            }
        }
    }

    /**
     * 修改文档
     *
     * @param indexName
     * @param documentId
     * @param map
     */
    @Override
    public void update(String indexName, String indexType, String documentId, Map<String, Object> map) {
        UpdateRequest updateRequest = new UpdateRequest(indexName, indexType, documentId).doc(map);
        try {
            UpdateResponse updateResponse = client.update(updateRequest, RequestOptions.DEFAULT);
            if (updateResponse.getResult() == DocWriteResponse.Result.CREATED) {
                logger.debug("此文档初次创建");
            } else if (updateResponse.getResult() == DocWriteResponse.Result.UPDATED) {
                logger.debug("文档更新操作成功");
            } else if (updateResponse.getResult() == DocWriteResponse.Result.NOOP) {
                logger.error("文档更新失败！");
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 异步：查询并批量修改文档
     *
     * @param indexName
     */
    @Override
    public BulkByScrollResponse updateByQuery(String indexName, QueryBuilder queryBuilder, Script script) {
        UpdateByQueryRequest updateRequest = new UpdateByQueryRequest(indexName);
        updateRequest.setQuery(queryBuilder);
        updateRequest.setScript(script);
        updateRequest.setConflicts("proceed");
        updateRequest.setBatchSize(1000);
        updateRequest.setSlices(5);
        updateRequest.setScroll(TimeValue.timeValueMinutes(60));
        updateRequest.setTimeout(TimeValue.timeValueMinutes(60));
        updateRequest.setRefresh(true);
        log.info("=====>筛选条件为：{}",queryBuilder.toString());
        log.info("=====>更新配置为：{}",script.toString());
//        ActionListener<BulkByScrollResponse> listener = new ActionListener<BulkByScrollResponse>() {
//            @Override
//            public void onResponse(BulkByScrollResponse bulkResponse) {
//                logger.info("=====>customerId=[{}]执行结果如下：",cid);
//                logger.info("=====>customerId=[{}]执行总时间：{}",cid,bulkResponse.getTook());
//                logger.info("=====>customerId=[{}]获取文档总数量：{}",cid,bulkResponse.getTotal());
//                logger.info("=====>customerId=[{}]更新文档成功数量：{}",cid,bulkResponse.getUpdated());
//                logger.info("=====>customerId=[{}]版本冲突的数量：{}",cid,bulkResponse.getVersionConflicts());
//                logger.info("=====>customerId=[{}]已执行的批次数量：{}",cid,bulkResponse.getBatches());
//                //ONE：标记成功
//                staticCustomerMapper.updIsMark(PublicConstant.ONE,cid,bulkResponse.getUpdated());
//            }
//            @Override
//            public void onFailure(Exception e) {
//                e.printStackTrace();
//            }
//        };
//        client.updateByQueryAsync(updateRequest, RequestOptions.DEFAULT,listener);
        try {
            BulkByScrollResponse updateResponse = client.updateByQuery(updateRequest, RequestOptions.DEFAULT);
            logger.info("=====>[{}]批量更新数量为：{}",script,updateResponse.getUpdated());
            return updateResponse;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 批量新增文档
     *
     * @param indexName
     * @param map
     */
    @Override
    public void bulkInsert(String indexName, String indexType, Map<String, Map<String, Object>> map) {
        BulkRequest bulkRequest = new BulkRequest();
        map.forEach((k, v) -> bulkRequest.add(new IndexRequest(indexName, indexType, k).source(v)));
        try {
            BulkResponse bulkResponse = client.bulk(bulkRequest, RequestOptions.DEFAULT);
            if (bulkResponse.hasFailures()) {
                for (BulkItemResponse response : bulkResponse.getItems()) {
                    if (response.isFailed()) {
                    }
                }
            }
//                throw new EngineException(999, "批量插入失败！");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * es执行批量新增或者更新
     *
     * @param indexName
     * @param map
     */
    @Override
    public void bulkUpdate(String indexName, String indexType, Map<String, Map<String, Object>> map) {
        BulkRequest bulkRequest = new BulkRequest();
        if (map == null || map.size() == 0) {
            return;
        }
        map.forEach((k, v) -> {
            String routing = StringUtils.isNoneEmpty(k) && k.length() >= 19 ? k.substring(11, 19) : "";
            bulkRequest.add(new UpdateRequest(indexName, k).doc(v).routing(routing));
            // bulkRequest.add(new UpdateRequest(indexName, k).doc(v).routing(""));
        });
        try {
            BulkResponse bulkResponse = client.bulk(bulkRequest, RequestOptions.DEFAULT);
            if (bulkResponse.hasFailures()) {
                Map<String, Map<String, Object>> retryMap = new HashMap<>(map.size());
                for (BulkItemResponse response : bulkResponse.getItems()) {
                    if (response.isFailed()) {
                        String id = response.getFailure().getId();
                        retryMap.put(id, map.get(id));
                    }
                }
                retryUpdate(indexName, retryMap, 0, indexType);
            }
        } catch (Exception e) {
            logger.info("retryUpdate发生异常："+e);
            retryUpdate(indexName, map, 0, indexType);
        }
    }

    private static final int MAX_RETRY_COUNT = 5;

    private void retryUpdate(String indexName, Map<String, Map<String, Object>> map, Integer count, String indexType) {
        logger.info("\nretryUpdate---------\n");
        if (count < MAX_RETRY_COUNT) {
            BulkRequest bulkRequest = new BulkRequest();
            map.forEach((k, v) -> {
                // String routing = StringUtils.isNoneEmpty(k) ? k.substring(11) : "";
                // bulkRequest.add(new UpdateRequest(indexName, k).doc(v).routing(routing));
                bulkRequest.add(new UpdateRequest(indexName, k).doc(v));
            });
            try {
                count++;
                BulkResponse bulkResponse = client.bulk(bulkRequest, RequestOptions.DEFAULT);
                if (bulkResponse.hasFailures()) {
                    Map<String, Map<String, Object>> retryMap = new HashMap<>(map.size());
                    for (BulkItemResponse response : bulkResponse.getItems()) {
                        if (response.isFailed()) {
                            String id = response.getFailure().getId();
                            retryMap.put(id, map.get(id));
                        }
                    }
                    retryUpdate(indexName, retryMap, count, indexType);
                }
            } catch (Exception e) {
                logger.error("retryUpdate发生异常:", e);
                retryUpdate(indexName, map, count, indexType);
            }
        }
    }

    /**
     * es执行批量删除
     *
     * @param indexName
     * @param map
     */
    @Override
    public void bulkUpdateWithDel(String indexName, String indexType, Map<String, String> map) {
        BulkRequest bulkRequest = new BulkRequest();
        map.forEach((k, v) -> bulkRequest.add(new UpdateRequest(indexName, indexType, k).script(new Script(v))));
        try {
            client.bulk(bulkRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 根据indexName，docId搜索文档
     *
     * @param indexName indexName
     * @param docId     docId
     */
    @Override
    public void searchDoc(String indexName, String indexType, String docId) {
        MultiGetRequest multiGetRequest = new MultiGetRequest();
        GetRequest request = new GetRequest(indexName, indexType, docId);
        String[] includes = Strings.EMPTY_ARRAY;
        String[] excludes = Strings.EMPTY_ARRAY;
        FetchSourceContext fetchSourceContext = new FetchSourceContext(true, includes, excludes);
        request.fetchSourceContext(fetchSourceContext);
        GetResponse getResponse = null;
        try {
            try {
                getResponse = client.get(request, RequestOptions.DEFAULT);
            } catch (IOException e) {
                e.printStackTrace();
            }
        } catch (ElasticsearchException e) {
            if (e.status() == RestStatus.NOT_FOUND) {
                logger.error("没有找到该id的文档");
            }
            if (e.status() == RestStatus.CONFLICT) {
                logger.error("获取时版本冲突了，请在此写冲突处理逻辑！");
            }
            logger.error("获取文档异常", e);
        }
        if (getResponse != null) {
            String index = getResponse.getIndex();
            String type = getResponse.getType();
            String id = getResponse.getId();
            if (getResponse.isExists()) {
                logger.info("index:" + index + "  type:" + type + "  id:" + id);
                logger.info("\n");
                getResponse.getSource().forEach((k, v) -> {
                    logger.info("key========>>" + k);
                    logger.info("value=======>>" + v);
                });
            } else {
                logger.error("没有找到该id的文档");
            }
        }
    }

    @Override
    public void scrollSearch(String indexName, FileLoadModel fileLoadModel, IQueryCallback iQueryCallback) {
        try {
            SearchRequest searchRequest = new SearchRequest(indexName).source(new SearchSourceBuilder().query(
                    fileLoadModel.getQueryBuilder())
                                                                                                       .sort("_doc")
                                                                                                       .trackTotalHits(true)
                                                                                                       .fetchSource(
                                                                                                               fileLoadModel
                                                                                                                       .getIncludes(),
                                                                                                               fileLoadModel
                                                                                                                       .getExcludes())
                                                                                                       .size(1000));
            Scroll scroll = new Scroll(TimeValue.timeValueMillis(600000));
            searchRequest.scroll(scroll);
            SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
            String scrollId = searchResponse.getScrollId();
            SearchHit[] hits = searchResponse.getHits().getHits();
            while (ArrayUtils.isNotEmpty(hits)) {
                iQueryCallback.resultListener(hits);
                SearchScrollRequest searchScrollRequest = new SearchScrollRequest(scrollId);
                searchScrollRequest.scroll(scroll);
                SearchResponse searchScrollResponse = client.searchScroll(searchScrollRequest, RequestOptions.DEFAULT);
                scrollId = searchScrollResponse.getScrollId();
                hits = searchScrollResponse.getHits().getHits();
            }
            ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
            clearScrollRequest.addScrollId(scrollId);
            client.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            e.printStackTrace();
            logger.error("滚动查询发生异常---->>>", e);
            throw new BulkLoadException(new ExceptionCode(999, e.getMessage(), "滚动查询发生异常"));
        }
    }

    private static final String MULTI_CUST_SIGN = "\\|";

    @SneakyThrows
    @Override
    public Map<String, String> multiSearchDoc(List<String> ids, String indexName, String[] includes) {
        MultiGetRequest multiGetRequest = new MultiGetRequest();
        FetchSourceContext fetchSourceContext = new FetchSourceContext(true, includes, Strings.EMPTY_ARRAY);
        // ids.forEach(id -> multiGetRequest.add(new MultiGetRequest.Item(indexName, BulkLoadConfig.ECIF_TYPE, id).routing(
        //         id.substring(11)).fetchSourceContext(fetchSourceContext)));
        ids.forEach(id -> multiGetRequest.add(new MultiGetRequest.Item(indexName, BulkLoadConfig.ECIF_TYPE, id)
                .fetchSourceContext(fetchSourceContext)));
        MultiGetResponse responses = client.multiGet(multiGetRequest, RequestOptions.DEFAULT);
        Map<String, String> result = new HashMap<>(responses.getResponses().length);
        for (MultiGetItemResponse response : responses.getResponses()) {
            if (null != response) {
                for (String include : includes) {
                    String custNo = null;
                    if (null != response.getResponse().getSourceAsMap()) {
                        custNo = response.getResponse().getSourceAsMap().get(include) + "";
                    }
                    if (StringUtils.isNotEmpty(custNo)) {
                        if (custNo.contains(MULTI_CUST_SIGN)) {
                            result.putAll(multiEcifHandler(custNo, response.getId()));
                        } else {
                            result.put(custNo, response.getId());
                        }
                    }
                }
            }
        }
        return result;
    }

    private Map<String, String> multiEcifHandler(String multiCustNO, String id) {
        Map<String, String> result = new HashMap<>(16);
        for (String s : multiCustNO.split(MULTI_CUST_SIGN)) {
            result.put(s, id);
        }
        return result;
    }

    @Override
    public void newScrollSearch(String indexName, FileLoadModel fileLoadModel, IQueryCallback iQueryCallback, int sliceId,
                             Lock lock, CountDownLatch countDownLatch) {
        try {
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder().query(fileLoadModel.getQueryBuilder())
                    .sort("_doc")
                    .slice(new SliceBuilder(sliceId,
                            ES_SLICE_MAX))
                    .fetchSource(fileLoadModel.getIncludes(),
                            fileLoadModel.getExcludes())
                    .size(5000);
            logger.info("索引是[{}], dataSet查询条件为----->>>>>{}", indexName, searchSourceBuilder);
            SearchRequest searchRequest = new SearchRequest(indexName).source(searchSourceBuilder);
            Scroll scroll = new Scroll(TimeValue.timeValueMillis(600000));
            searchRequest.scroll(scroll);
            StopWatch queryWatch = new StopWatch();
            StopWatch writeWatch = new StopWatch();
            queryWatch.start();
            SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
            queryWatch.stop();
            String scrollId = searchResponse.getScrollId();
            SearchHit[] hits = searchResponse.getHits().getHits();
            while (ArrayUtils.isNotEmpty(hits)) {
                writeWatch.start();
                try {
                    lock.tryLock(1, TimeUnit.MINUTES);
                    iQueryCallback.resultListener(hits);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    lock.unlock();
                }
                writeWatch.stop();
                logger.info("\n\n\n======> 查询出{}行数据\n        查询用时{}毫秒，\n        写入用时{}毫秒\n\n",
                        hits.length,
                        queryWatch.getTime(),
                        writeWatch.getTime(TimeUnit.MILLISECONDS));
                queryWatch.reset();
                writeWatch.reset();
                SearchScrollRequest searchScrollRequest = new SearchScrollRequest(scrollId);
                searchScrollRequest.scroll(scroll);
                queryWatch.start();
                SearchResponse searchScrollResponse = client.scroll(searchScrollRequest, RequestOptions.DEFAULT);
                queryWatch.stop();
                scrollId = searchScrollResponse.getScrollId();
                hits = searchScrollResponse.getHits().getHits();
            }
            ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
            clearScrollRequest.addScrollId(scrollId);
            client.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            logger.error("滚动查询发生异常---->>>", e);
            throw new BulkLoadException(new ExceptionCode(999, e.getMessage(), "滚动查询发生异常"));
        } finally {
            countDownLatch.countDown();
        }
    }
}

